package com.pwrd.sframe.core;

import com.pwrd.sframe.core.thread.RunnableCase;
import com.pwrd.sframe.core.thread.ThreadHandler;
import com.pwrd.sframe.support.exception.SysException;
import com.pwrd.sframe.support.function.Func;
import com.pwrd.sframe.support.log.Log;
import com.pwrd.sframe.core.obverser.RemoteCallObverser;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;

/**
 * @Auther syc.
 * @Date 2017/1/22 15:06.
 */
public class Port implements RunnableCase {

    private Node node;
    private String key;
    Map<Object, Service> services = new HashMap<>();
    private ThreadHandler thread;

    // 调用请求队列
    private Queue<Call> callsQ = new LinkedList<>();
    // 一次心跳处理的调用请求队列
    private Queue<Call> callsPulseQ = new LinkedList<>();

    // 对外调用请求队列
    private Queue<Call> outCallsQ = new LinkedList<>();


    public Port(String portKey) {
        this.key = portKey;
        thread = new ThreadHandler(this, key.toString());
    }

    public void addService(Object key, Service service) {
        services.put(key, service);
    }

    public void startup(Node node) {
        this.node = node;
        node.addPort(key, this);

        thread.startup();
    }

    @Override
    public void run() {
        pulse();
    }

    private final void pulse() {
        // 缓存本次心跳要处理的调用请求
        while (!callsQ.isEmpty()) {
            callsPulseQ.offer(callsQ.poll());
        }

        // 处理调用请求
        while (!callsPulseQ.isEmpty()) {
            pulseCall(callsPulseQ.poll());
        }

        // 下属service心跳
        pulseService();

        // port子类的心跳
        pulseOverride();

        // 发送调用
        while (!outCallsQ.isEmpty()) {
            outPulseCall(outCallsQ.poll());
        }

    }

    /**
     * 发送rpc调用
     * @param call
     */
    private void outPulseCall(Call call) {
        try{
            // 发送本地Node远程调用
            node.callHandle(call);
            // 发送其他Node远程调用
            // TODO
        } catch (Throwable e) {
            // 输出异常，服务器持续运行
            Log.core.error("Port: {} occur out call error, Call: {}", this.getKey(), call);
            e.printStackTrace();
        }
    }

    /**
     * 子类添加心跳操作
     */
    protected void pulseOverride() {
    }

    private void pulseService() {
        Service service = null;
        try{
            for (Service s : services.values()) {
                service = s;
                service.pulse();
            }
        } catch (Throwable e) {
            // 输出异常，服务器持续运行
            Log.core.error("Port: {} occur one Service pulse error, Call: {}", this.getKey(), service);
            e.printStackTrace();
        }
    }

    protected void pulseCall(Call call) {

        try {
            // 获取所需服务的实例
            Service service = services.get(call.toPoint.serviceKey);
            if(service == null) throw new SysException("Service:key={} is not belong to Port:key={}", call.toPoint.serviceKey, key);

            call.params[0] = service;

            // 远程调用方法
            callHandler(call);
        } catch (Throwable e) {
            // 输出异常，服务器持续运行
            Log.core.error("Port: {} occur one Call error, Call: {}", this.getKey(), call);
            e.printStackTrace();
        }
    }

    /**
     * 创建一个调用：本地port上所属服务调用外部服务
     *
     * @param toPoint
     * @param methodKey
     * @param params
     */
    public void call(CallInfo toPoint, String methodKey, Object[] params) {

        // 引用传递
        if (isSameNode(node.getKey(), toPoint.nodeKey)) {
            call(false, toPoint, methodKey, params);
        }
        // 值传递
        else {
            call(true, toPoint, methodKey, params);
        }
    }

    /**
     * 创建一个调用：本地port上所属服务调用外部服务
     * @param copy true：调用复制后发送
     * @param toPoint
     * @param methodKey
     * @param params
     */
    public void call(boolean copy, CallInfo toPoint, String methodKey, Object[] params) {

        Call call = new Call(copy, new CallInfo(node.getKey(), getKey()), toPoint, methodKey, params);

        // 不同节点调用不能引用传递，只能值传递
        if(!isSameNode(call)) {
            call.copy = true;
        }

        sendCall(call);
    }

    /**
     * 发送远程调用
     * @param call
     */
    public void sendCall(Call call) {
        // 发送到本地Node
        if(isSameNode(call)) {
            outCallsQ.add(call.copy? call.deepCall() : call);
        }
        // 发送其他Node
        else {
            // TODO
        }
    }

    public boolean isSameNode(String fromNodeKey, String toNodeKey) {
        boolean isSameNode = false;
        if(fromNodeKey.equals(toNodeKey)) {
            isSameNode = true;
        }
        return isSameNode;
    }

    public boolean isSameNode(Call call) {
        String fromNodeKey = call.fromPoint.nodeKey;
        String toNodeKey = call.toPoint.nodeKey;
        return isSameNode(fromNodeKey, toNodeKey);
    }

    public boolean isSamePort(Call call) {
        String fromPortKey = call.fromPoint.portKey;
        String toPortKey = call.toPoint.portKey;
        boolean isSamePort = false;
        if(isSameNode(call) && fromPortKey.equals(toPortKey)) {
            isSamePort = true;
        }
        return isSamePort;
    }

    // 处理本地port的调用
    public void callHandler(Call call) {
        String methodKey = call.methodKey;
        Object[] objects = call.params;
        Func fun = RemoteCallObverser.get(methodKey);
        fun.use(objects);
    }

    public String getKey() {
        return key;
    }

    /**
     * 接收Call请求
     * @param call
     */
    public void addCall(Call call) {
        callsQ.add(call);
    }
}
